import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Row;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;


public class HBaseTest {
	public static void main(String[] args) throws IOException{
		
		//hbase-site.xml(hbase-default.xml) 파일 loading
		Configuration conf = HBaseConfiguration.create();
		
		if(args.length < 1){
			System.out.println("Type 1(Creat), 2(Insert), 3(Select), 4(Delete), 5(Drop), 6(Batch) :");
			System.exit(0);
		}
		
		/*
		 * 
		//스레드 안전(thread-safe)하지 않음.
		HTablePool pool = new HTablePool(conf, 5); //Integer.MAX_VALUE
		
		//Pool에 설정한 최대 개수가 넘는 인스턴스도 문제없이 생성
		HTableInterface[] tables = new HTableInterface[10];  
		//HTableInterfaceFactory 로 Customizing 가능
		//createHTableInterface, releaseHTableInterface(버퍼를 비우거나 flushComits()) 구현함
		//
		//
		for(int n = 0; n < 10; n++){
			tables[n] = pool.getTable("testtable");
			System.out.println(Bytes.toString(tables[n].getTableName()));
		}
		
		//Pool 에 반환 (5개만)
		for(int n = 0; n < 10; n++){
			tables[n].close();		
		}
	
		//보유중인 테이블 참조를 해제하려면(일정 시간이 지나면 참조 테이블 close해야함)
		pool.closeTablePool("testtable"); 
		*/
		
		int type = Integer.parseInt(args[0]);
		switch (type) {
		case 1:
			//Create Table 
			HBaseAdmin createHbase = new HBaseAdmin(conf);
			HTableDescriptor createDesc = new HTableDescriptor("WIZARD_TABLE");
			
			//createDesc.setMaxFileSize(maxFileSize); //각 컬럼패밀리에 속한 모든 파일에 해당하는 저장소의 최대크기(기본값 256MB) 
			//createDesc.setMemStoreFlushSize(memstoreFlushSize); //플러시 작업(디스트 저장 파일로 쓰는 작업)이 일어나는 시기를 바이트 단위로 제어(기본값 64MB)
			//createDesc.setDeferredLogFlush(isDeferredLogFlush); 
			// WAL의 데이터를 디스크에 저장할 때 지연된(deferred)로그[변경사항이 서버에 전송될때 마다 곧바로 실행되는게 아니고 백그라운드 프로세스로 전환] 플러시를 사용할지 안할지 결정(기본값 false)
			// 
		
			//Column Family는 너무 많이 생성하면 안된다.
			//Column Family이름이 곧 파일 경로로 사용되므로 파일명 규칙에 부합(인쇄가능한문자)해야함.
			HColumnDescriptor createFam1 = new HColumnDescriptor("ColFam1".getBytes()); 
			HColumnDescriptor createFam2 = new HColumnDescriptor("ColFam2".getBytes());
			
			//createFam1.setMaxVersions(maxVersions); //최대 개수를 초과하는 버전값을 삭제
			//createFam1.setBlocksize(s); //읽기 또는 스캔 연산 시에 로드되는 더 작은 블록으로 구분(기본값 64KB)
			//createFam1.setBlockCacheEnabled(blockCacheEnabled); 
			//데이터 전체 블록을 읽어들이기 때문에 이후의 읽기에서 디스크 접근이 필요하지 않도록 인메모리 캐시에 유지.(기본값 true)
	        //순차적으로 일기만 하고  재사용하지 않는 경우라면 false로 설정
			//createFam1.setInMemory(inMemory); //메모리에 로드된 컬럼패밀리의 모든 블록을 계속 유지 보장
			
			createDesc.addFamily(createFam1);
			createDesc.addFamily(createFam2);
			
			createHbase.createTable(createDesc);
			//데이터 갱신 --> WAL(Write-Ahead Log{Commit Log}) --> Memstore(메모리) --> HFile로 Flush
			try {
				createHbase.flush("WIZARD_TABLE");
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			
			createHbase.close();
			break;

		case 2:
			//Insert Data (TimeStamp 는 HBasse의 logic에 맡기는것이 효율적)
			HTable insertHbase = new HTable(conf, "WIZARD_TABLE");
			//클라이언트 측 버퍼는 기본적으로 비활성화 되어있기때문에 활성화 시킨다.(매번 RPC 호출 최소화)
			insertHbase.setAutoFlush(false);
			Put put = new Put(Bytes.toBytes("rowkey1"));
			
			if(put.has(Bytes.toBytes("ColFam1"), Bytes.toBytes("Qual1"))){
				System.out.println("[" + Bytes.toString(put.getRow()) + "]키의 Column Family 와 Qualifier가 존재한다.");
			}else{
				System.out.println("[" + Bytes.toString(put.getRow()) + "]키의 Column Family 와 Qualifier가 존재안한다.");
			}
			
			//데이터를 WAL에 저장하면 취소
			if(put.getWriteToWAL()){ 
				put.setWriteToWAL(false);
			}
			
			put.add(Bytes.toBytes("ColFam1"), Bytes.toBytes("Qual1"), Bytes.toBytes("val1"));
			put.add(Bytes.toBytes("ColFam1"), Bytes.toBytes("Qual1"), Bytes.toBytes("val3"));
			put.add(Bytes.toBytes("ColFam1"), Bytes.toBytes("Qual2"), Bytes.toBytes("val2"));
			
			//List 로 Insert 가능
			//List<Put> puts = new ArrayList<Put>();
			//puts.add(put);
			//insertHbase.put(puts);
			insertHbase.put(put);
			
			/*
			 * 버퍼가 비워지는 경우
			 * 1: Explicit -> insertHbase.flushCommits();
			 * 2: Implicit -> getWriteBufferSize() 의 크기에 도달할 경우[기본 2MB(2,097,152)] -> setWriteBufferSize()로 설정가능
			 *                HTable이 close()될 경우
			 * 
			 * hbase-site.xml에서 버퍼제한 늘릴 수 있음
			 * <property>
			 *     <name>hbase.client.write.buffer</name>
			 *     <value>20971520</value> // 20MB
			 * </property>
			 */
			insertHbase.close();
			
			
			
			/*
			 *동일한 Row에사 확인후 Insert가 가능하다.(테스트 후 주석 처리)
			 */
			
			
		    /*
		    Put put1 = new Put(Bytes.toBytes("row1"));
		    put1.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"),
		      Bytes.toBytes("val1")); // 새로운 Put 생성

		    boolean res1 = insertHbase.checkAndPut(Bytes.toBytes("row1"),
		      Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), null, put1); // 해당 컬럼이 존재하는지 확인
		    System.out.println("Put applied: " + res1); // 해당 컬럼이 없으므로 True

		    boolean res2 = insertHbase.checkAndPut(Bytes.toBytes("row1"),
		      Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), null, put1); 
		    System.out.println("Put applied: " + res2); // 해당 컬럼이 없으므로 False

		    Put put2 = new Put(Bytes.toBytes("row1"));
		    put2.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual2"),
		      Bytes.toBytes("val2")); // 새로운 Put 생성

		    boolean res3 = insertHbase.checkAndPut(Bytes.toBytes("row1"),
		      Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), //이전데이터가 저장된 경우에만 새로운 데이터 저장
		      Bytes.toBytes("val1"), put2);
		    System.out.println("Put applied: " + res3); //해당 컬럼의 데이터가 존재하므로 True

		    Put put3 = new Put(Bytes.toBytes("row2"));
		    put3.add(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"),
		      Bytes.toBytes("val3")); // 다른 RowKey로 새로운 Put 생성

		    boolean res4 = insertHbase.checkAndPut(Bytes.toBytes("row1"),
		      Bytes.toBytes("colfam1"), Bytes.toBytes("qual1"), // 다른 Row를 확인 하므로 Exception 발생
		      Bytes.toBytes("val1"), put3);
		    System.out.println("Put applied: " + res4); //Exception
		    */
			
			
			break;
			
		case 3:
	
			//특정 Row Select (테스트 후 주석처리)
			/*
		    HTable table = new HTable(conf, "WIZARD_TABLE");

		    Get get = new Get(Bytes.toBytes("rowkey1")); // Row 설정
		    get.addColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1")); // Column Family, Qualifier 설정 
		    
		    if(table.exists(get)){
		    	//해당하는 조건의 데이터가 존재하는지 확인 -> 일반 get 함수와 같이 모든 리전 서버 탐색, 단지 데이터를 전송하는 과정만 회피
		    	  [아주 큰 컬럼의 존재 여부를 확인할 때 매우 유용]
		    }
		    
		    Result result = table.get(get); 

		    byte[] val = result.getValue(Bytes.toBytes("colfam1"),
		      Bytes.toBytes("qual1")); // 가장 최신 버전데이터 조회[setMaxVersions로 수정가능]
		    */
			
			
		    //Select Data
			HTable selectHbase = new HTable(conf, "WIZARD_TABLE");
			
	        Scan s =new Scan();
	        //조회조건설정예1 : 해당 Key부터 조회
	        s.setStartRow(Bytes.toBytes("rowkey1"));
	        //조회조건설정예2 : 같은 Column Family와 Qualifier에 데이터가 저정될 경우 가장 최신에 저장된 데이터만 Select --> 3개의 데이터를 조회해 오는것으로 수정
	        s.setMaxVersions(3); 
	        //조회조건설정예3 : Key값에 "rowkey"가 존재하면 조회
	        //setTimeRange
	        RowFilter rf = new RowFilter(CompareOp.EQUAL, new SubstringComparator("rowkey"));
	        s.setFilter(rf);
	        
	        ResultScanner rs = selectHbase.getScanner(s);
	        //Get과 비슷한 연산으로 변환, 결국 next() 메소드를 실행하여 로우당 하나씩의 RPC가 수행
	        //rs.next(return count)도 결국 클라이언트에서 해당 Count만큼 메소드를 반복처리하는 것임.
	        for (Result r : rs) {
	            for (KeyValue kv : r.raw()) {
	                System.out.println("row:" + new String(kv.getRow()) +"");
	                System.out.println("family:" + new String(kv.getFamily()) +":");
	                System.out.println("qualifier:" + new String(kv.getQualifier()) +"");
	                System.out.println("value:" + new String(kv.getValue()));
	                System.out.println("timestamp:" + kv.getTimestamp() +"");
	                System.out.println("-------------------------------------------");
	            }
	        }
	        
	        /*
	         * Row단위 Cache
	         * 테이블 단위의 Caching : setScannerCaching(int ), getScannerCaching()
	         * 스캔 단위의 Caching :setCaching(int ), getCaching()
	         * 
	         *  hbase-site.xml 속성 변경
	         *  <property>
	         *      <name>hbase.client.scanner.caching</name>
	         *      <value>10</value>
	         *  </property>
	         *  
	         * Column 단위 Cache (Column이 많을 경우 필요)
	         * setBatch(int ), getBatch()
	         * 
	         * Ex.) 로우당 20개의 칼럼을 가지고 각 컬럼에는 버전이 하나뿐이 모두 200개의 컬럼이 존재할 경우
	         * setCaching(2), setBatch(10) 일경우 200 / 10 = 20, 즉 Result 20번 --> 2번의 Caching 으로 RPC는 10번 만 호출 
	         */
	        
	        //잘못된 동작으로 너무 오랫동안 리소스를 점유하는 클라이언트로 부터 보호하기 위해 hbase-site.xml 설정
	        //<property>
	        //	<name>hbase.regionserver.lease.period</name>
	        //	<value>120000</value>
	        //</property>
	        rs.close();
	        selectHbase.close();
	        
	        //리스트 Get Select (테스트 완료 후 주석처리)
	        /*
	        byte[] cf1 = Bytes.toBytes("colfam1");
	        byte[] qf1 = Bytes.toBytes("qual1");
	        byte[] qf2 = Bytes.toBytes("qual2"); 
	        byte[] row1 = Bytes.toBytes("row1");
	        byte[] row2 = Bytes.toBytes("row2");

	        List<Get> gets = new ArrayList<Get>();  

	        Get get1 = new Get(row1);
	        get1.addColumn(cf1, qf1);
	        gets.add(get1);

	        Get get2 = new Get(row2);
	        get2.addColumn(cf1, qf1); 
	        gets.add(get2);

	        Get get3 = new Get(row2);
	        get3.addColumn(cf1, qf2);
	        gets.add(get3);

	        Result[] results = selectHbase.get(gets); 

	        System.out.println("First iteration...");
	        for (Result result : results) {
	          String row = Bytes.toString(result.getRow());
	          System.out.print("Row: " + row + " ");
	          byte[] val = null;
	          if (result.containsColumn(cf1, qf1)) { 
	            val = result.getValue(cf1, qf1);
	            System.out.println("Value: " + Bytes.toString(val));
	          }
	          if (result.containsColumn(cf1, qf2)) {
	            val = result.getValue(cf1, qf2);
	            System.out.println("Value: " + Bytes.toString(val));
	          }
	        }

	        System.out.println("Second iteration...");
	        for (Result result : results) {
	          for (KeyValue kv : result.raw()) {
	            System.out.println("Row: " + Bytes.toString(kv.getRow()) +
	              " Value: " + Bytes.toString(kv.getValue()));
	          }
	        }
	        */
			break;
			
			
		case 4:
			
		    //delete Data
			HTable deleteHbase = new HTable(conf, "WIZARD_TABLE");
			
			//해당 Row 삭제
			List<Delete> list =new ArrayList<Delete>();
            Delete d1 =new Delete("rowkey1".getBytes());
            //d1.setTimestamp(timestamp) //해당 TimeStamp 와 그이전의 데이터 삭제 (deleteColumns, deleteFamily 설정 안 할 경우 적용됨)
            //d1.deleteColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual1")); //모든 버전 삭제
            //d1.deleteColumn(Bytes.toBytes("colfam1"), Bytes.toBytes("qual3"), timestamp); //특정 버전 및 이전 버전 삭제
            //d1.deleteFamily(Bytes.toBytes("colfam1")); // 모든 버전의 Column Family 삭제
            //d1.deleteFamily(Bytes.toBytes("colfam1"), timestamp); //특정 버전및 이전 Column Family 삭제
            
            list.add(d1);
            deleteHbase.delete(list);
            //deleteHbase.delete(d1); 특정 Row만 삭제
            deleteHbase.close();
            
            //같은 Row의 Column은 Check 연산 가능
            /*
		    Delete delete1 = new Delete(Bytes.toBytes("row1"));
		    delete1.deleteColumns(Bytes.toBytes("colfam1"), Bytes.toBytes("qual3")); 

		    boolean res1 = deleteHbase.checkAndDelete(Bytes.toBytes("row1"),
		      Bytes.toBytes("colfam2"), Bytes.toBytes("qual3"), null, delete1); //해당 Column이 존재하는지 확인 후 연산 수행
		    System.out.println("Delete successful: " + res1); // 현재 존재하므로 False
			*/
		    
			break;
			
		case 5:
		    // ���̺� drop
            HBaseAdmin admin =new HBaseAdmin(conf);
            admin.disableTable("WIZARD_TABLE");
            admin.deleteTable("WIZARD_TABLE");
            admin.close();
			break;
		
		case 6:
			//일괄로 모든 작업 처리
			/*
		    private final static byte[] ROW1 = Bytes.toBytes("row1");
		    private final static byte[] ROW2 = Bytes.toBytes("row2");
		    private final static byte[] COLFAM1 = Bytes.toBytes("colfam1"); // co BatchExample-1-Const Use constants for easy reuse.
		    private final static byte[] COLFAM2 = Bytes.toBytes("colfam2");
		    private final static byte[] QUAL1 = Bytes.toBytes("qual1");
		    private final static byte[] QUAL2 = Bytes.toBytes("qual2");
		    
		    HTable table = new HTable(conf, "WIZARD_TABLE");

		    // vv BatchExample
		    List<Row> batch = new ArrayList<Row>(); 

		    Put put = new Put(ROW2);
		    put.add(COLFAM2, QUAL1, Bytes.toBytes("val5")); 
		    batch.add(put);

		    Get get1 = new Get(ROW1);
		    get1.addColumn(COLFAM1, QUAL1);
		    batch.add(get1);

		    Delete delete = new Delete(ROW1);
		    delete.deleteColumns(COLFAM1, QUAL2);
		    batch.add(delete);

		
		    Object[] results = new Object[batch.size()]; 
		    try {
		      table.batch(batch, results);
		    } catch (Exception e) {
		      System.err.println("Error: " + e); 
		    }

		    for (int i = 0; i < results.length; i++) {
		      System.out.println("Result[" + i + "]: " + results[i]); 
		      //null : 연산이 서버와 통신에 실패
		      //빈 result : put 또는 delete 연산에 성공
		      //result : get 연산 성공, 로우나 컬럼이 없는 경우 빈 result
		      //throwable : 서버에서 예외를 던진 경우, 클라이언트 측에 그대로 반환  
		    }*/
		    
			break;
		default:
			break;
		}

		

        
	}
}
